#include <iostream>
#include <cmath>
#include <limits>
#include <vector>
#include <map>
#include <cstdlib>
#include "SharedArray.h"
#include "Column.h"
#include "types.h"
#include "reordering.h"
#include "ConstantExpressionNode.h"
#include "ColumnExpressionNode.h"
#include "UnaryExpressionNode.h"
#include "BinaryExpressionNode.h"
#include "IfExpressionNode.h"
#include "operators.h"
#include "HashTable.h"
#include "server.h"
#include "ScanOperationNode.h"
#include "ComputeOperationNode.h"
#include "FilterOperationNode.h"
#include "GroupByOperationNode.h"
#include "AggregationData.h"
#include "Consumer.h"
#include "Parser.h"

using namespace std;

static const double EPSILON = 1e-5;

#define testThat(X) if (!(X)) { \
	cout << "Test " << #X << " in " << __FILE__ << ":" << __LINE__ << " failed!\n"; \
	exit(1); \
}

class OneTwoServer: public Server {
	public:
		OneTwoServer(int n): _n(n) {}

		virtual int GetDoubles(int column_index, int number, double* destination) {
			int pulled_number = setPulledAmount(column_index, number);

			for (int i = 0; i < pulled_number; ++i) {
				destination[i] = 1.0 + (i % 2) * 1.0;
			}

			return pulled_number;
		}

		virtual int GetInts(int column_index, int number, int32_t* destination) {
			int pulled_number = setPulledAmount(column_index, number);

			for (int i = 0; i < pulled_number; ++i) {
				destination[i] = 1 + (i % 2) * 1;
			}

			return pulled_number;
		}

		virtual int GetByteBools(int column_index, int number, bool* destination) {
			int pulled_number = setPulledAmount(column_index, number);

			for (int i = 0; i < pulled_number; ++i) {
				destination[i] = (i % 2 == 1);
			}

			return pulled_number;
		}

		virtual int GetBitBools(int column_index, int number, char* destination) {
			int pulled_number = setPulledAmount(column_index, number);

			for (int i = 0; i < (pulled_number + 7) / 8; ++i) {
				destination[i] = 0x22 + 0x88;
			}

			return pulled_number;
		}

		virtual void ConsumeDoubles(int column_index, int number, const double* destination) {
			for (int i = 0; i < number; ++i) {
				consumedDoubles[column_index].push_back(destination[i]);
			}
		}

		virtual void ConsumeInts(int column_index, int number, const int32_t* destination) {
			for (int i = 0; i < number; ++i) {
				consumedInts[column_index].push_back(destination[i]);
			}
		}

		virtual void ConsumeByteBools(int column_index, int number, const bool* destination) {
			for (int i = 0; i < number; ++i) {
				consumedBools[column_index].push_back(destination[i]);
			}
		}

		virtual void ConsumeBitBools(int column_index, int number, const char* destination) {
			for (int i = 0; i < number / 8; ++i) {
				for (char bit = 1; bit <= 128; bit *= 2) {
					consumedBools[column_index].push_back(destination[i] & bit);
				}
			}

			int leftBools = number % 8;
			int i = (number + 7) / 8;
			for (char bit = 1; bit <= 128; bit *= 2) {
				if (leftBools == 0)
					break;

				consumedBools[column_index].push_back(destination[i] & bit);
				--leftBools;
			}
		}

		// map: column_index -> vector of consumed data
		map<int, vector<double> > consumedDoubles;
		map<int, vector<int32_t> > consumedInts;
		map<int, vector<bool> > consumedBools;

	private:
		int setPulledAmount(int column_index, int requested) {
			int available;

			if (_columnN.find(column_index) != _columnN.end()) {
				available = _columnN[column_index];
			} else {
				available = _n;
			}

			int pulled_number = requested;
			if (available < requested)
				pulled_number = available;

			_columnN[column_index] = available - pulled_number;

			return pulled_number;
		}

		map<int, int> _columnN;
		int _n;
};

int main() {
	{
		SharedArray<int32_t> a;
		a = SharedArray<int32_t>(2);
		a[0] = 2;
		a[1] = 1;
		testThat(a[0] == 2);
		{
			SharedArray<int32_t> b(a);
			testThat(b[1] == 1);
			b[0] = 3;
			SharedArray<int32_t> c(b);
			testThat(c.references() == 3);
			testThat(c == a);
		}
		testThat(a[0] == 3);
		testThat(a.size() == 2);
		testThat(a.references() == 1);
		{
			SharedArray<int32_t> c(1);
			c[0] = 9;
			testThat(a != c);
			a = c;
			testThat(a == c);
			testThat(a.references() == 2);
			c = a.deepCopy();
			testThat(a.references() == 1);
			testThat(c.references() == 1);
		}
		testThat(a[0] == 9);
		testThat(a.size() == 1);
		testThat(a.references() == 1);

		{
			Column col = Column(a);
			testThat(col.references() == 2);

			{
				SharedArray<int32_t> b = col.toSharedArray<int32_t>();
				testThat(b == a);
				testThat(b.references() == 3);
			}

			testThat(col.references() == 2);
		}

		testThat(a.references() == 1);
	}
	{
		const int N = 8 * 1024 + 16;

		SharedArray<Column> data(5);
		SharedArray<int32_t> data0(N);
		SharedArray<int32_t> data1(N);
		SharedArray<double> data2(N);
		SharedArray<double> data3(N);
		SharedArray<bool> data4(N);
		for (int i = 0; i < N; ++i) {
			data0[i] = i;
			data1[i] = i % 16;
			data2[i] = (double) i / 2.0;
			data3[i] = 1 / (double) i;
			data4[i] = (i % 2 == 0);
		}
		data[0] = Column(data0);
		data[1] = Column(data1);
		data[2] = Column(data2);
		data[3] = Column(data3);
		data[4] = Column(data4);

		{
			ExpressionNode* root = new ConstantExpressionNode<int32_t>(42);
			Column result = root->pull(data);
			testThat(result.type() == INT);
			testThat(result.size() == N);
			testThat(result.at<int32_t>(N / 2) == 42);
			delete root;
		}

		{
			ExpressionNode* root = new ColumnExpressionNode(4);
			Column result = root->pull(data);
			testThat(result.type() == BOOL);
			testThat(result.size() == N);
			testThat(result.at<bool>(16) == true);
			testThat(result.at<bool>(31) == false);
			delete root;
		}

		{
			// if col0 - 8 > 2 * col1 and not col4 then col2 else col3
			ExpressionNode* col0 = new ColumnExpressionNode(0);
			ExpressionNode* eight = new ConstantExpressionNode<int32_t>(8);
			ExpressionNode* subtract = new BinaryExpressionNode<SubtractOperator>(col0, eight);
			ExpressionNode* two = new ConstantExpressionNode<int32_t>(2);
			ExpressionNode* col1 = new ColumnExpressionNode(1);
			ExpressionNode* multiply = new BinaryExpressionNode<MultiplyOperator>(two, col1);
			ExpressionNode* greater = new BinaryExpressionNode<GreaterOperator>(subtract, multiply);
			ExpressionNode* col4 = new ColumnExpressionNode(4);
			ExpressionNode* not_ = new UnaryExpressionNode<NotOperator>(col4);
			ExpressionNode* and_ = new BinaryExpressionNode<AndOperator>(greater, not_);
			ExpressionNode* col2 = new ColumnExpressionNode(2);
			ExpressionNode* col3 = new ColumnExpressionNode(3);
			ExpressionNode* ifThenElse = new IfExpressionNode(and_, col2, col3);

			ExpressionNode* root = ifThenElse;
			Column result = root->pull(data);
			testThat(result.type() == DOUBLE);
			testThat(result.size() == N);
			testThat(fabs(result.at<double>(1) - 1.0) < EPSILON);
			testThat(fabs(result.at<double>(101) - 50.5) < EPSILON);
			testThat(fabs(result.at<double>(102) - 1 / 102.0) < EPSILON);
			testThat(fabs(result.at<double>(19) - 9.5) < EPSILON);
			testThat(result.at<double>(0) == numeric_limits<double>::infinity());
			delete root;
		}
	}
	{
		char zero = 0;
		HashTable h(3, 1, &zero);
		hash_t hashes[5] = {1,2,3,1,2};
		SharedArray<char*> res = h.get("k10k20k30k11k20", hashes, 5);
		for (int i = 0; i < 5; ++i)
			testThat(*res[i] == zero);
		for (int i = 0; i < 5; ++i)
			++*res[i];
		testThat(*res[0] == 1);
		testThat(*res[1] == 2);
		testThat(*res[2] == 1);
		testThat(*res[3] == 1);
		testThat(res[4] == res[1]);

		SharedArray<KeyValue> kv = h.getEverything();
		testThat(kv.size() == 4);
	}
	{
		SharedArray<Type> types(3);
		types[0] = INT;
		types[1] = DOUBLE;
		types[2] = BOOL;
		SharedArray<int> columns(3);
		for (int i = 0; i < 3; ++i)
			columns[i] = i;

		Server* server = new OneTwoServer(PULL_AMOUNT + PULL_AMOUNT / 2);
		testThat(PULL_AMOUNT % 2 == 0);
		ScanOperationNode scan(types, columns, server);

		SharedArray<Column> result = scan.pull();

		testThat(result.size() == 3);
		for (int i = 0; i < 3; ++i)
			testThat(result[i].size() == PULL_AMOUNT);
		testThat(result[0].type() == INT);
		testThat(result[1].type() == DOUBLE);
		testThat(result[2].type() == BOOL);
		testThat(result[0].at<int32_t>(20) == 1);
		testThat(result[1].at<double>(31) == 2.0);
		testThat(result[2].at<bool>(120) == false);

		result = scan.pull();

		testThat(result.size() == 3);
		for (int i = 0; i < 3; ++i)
			testThat(result[i].size() == PULL_AMOUNT / 2);
		testThat(result[0].type() == INT);
		testThat(result[1].type() == DOUBLE);
		testThat(result[2].type() == BOOL);
		testThat(result[0].at<int32_t>(20) == 1);
		testThat(result[1].at<double>(31) == 2.0);
		testThat(result[2].at<bool>(120) == false);

		delete server;
	}
	{
		SharedArray<Type> types(3);
		types[0] = INT;
		types[1] = DOUBLE;
		types[2] = BOOL;
		SharedArray<int> columns(3);
		for (int i = 0; i < 3; ++i)
			columns[i] = i;

		Server* server = new OneTwoServer(PULL_AMOUNT + PULL_AMOUNT / 2);
		testThat(PULL_AMOUNT % 2 == 0);
		OperationNode* scan = new ScanOperationNode(types, columns, server);

		SharedArray<ExpressionNode*> expressionRoots(3);
		expressionRoots[0] = new ConstantExpressionNode<int32_t>(666);
		expressionRoots[1] = new ColumnExpressionNode(1);
		ExpressionNode* ifTrueThen3Else1 = new IfExpressionNode(new ColumnExpressionNode(2), new ConstantExpressionNode<int32_t>(3), new ConstantExpressionNode<int32_t>(1));
		expressionRoots[2] = new BinaryExpressionNode<AddOperator>(new ColumnExpressionNode(0), ifTrueThen3Else1);

		ComputeOperationNode compute(scan, expressionRoots);

		SharedArray<Column> result = compute.pull();

		testThat(result.size() == 3);
		for (int i = 0; i < 3; ++i)
			testThat(result[i].size() == PULL_AMOUNT);
		testThat(result[0].type() == INT);
		testThat(result[1].type() == DOUBLE);
		testThat(result[2].type() == INT);
		testThat(result[0].at<int32_t>(30) == 666);
		testThat(result[1].at<double>(41) == 2.0);
		testThat(result[2].at<int32_t>(110) == 2);
		testThat(result[2].at<int32_t>(111) == 5);

		result = compute.pull();

		testThat(result.size() == 3);
		for (int i = 0; i < 3; ++i)
			testThat(result[i].size() == PULL_AMOUNT / 2);
		testThat(result[0].type() == INT);
		testThat(result[1].type() == DOUBLE);
		testThat(result[2].type() == INT);
		testThat(result[0].at<int32_t>(30) == 666);
		testThat(result[1].at<double>(41) == 2.0);
		testThat(result[2].at<int32_t>(110) == 2);
		testThat(result[2].at<int32_t>(111) == 5);

		delete server;
	}
	{
		SharedArray<Type> types(3);
		types[0] = INT;
		types[1] = DOUBLE;
		types[2] = BOOL;
		SharedArray<int> columns(3);
		for (int i = 0; i < 3; ++i)
			columns[i] = i;

		Server* server = new OneTwoServer(3 * PULL_AMOUNT);
		testThat(PULL_AMOUNT % 2 == 0);
		OperationNode* scan = new ScanOperationNode(types, columns, server);

		ExpressionNode* filterCondition = new ColumnExpressionNode(2);
		FilterOperationNode filter(scan, filterCondition);

		SharedArray<Column> result = filter.pull();

		testThat(result.size() == 3);
		for (int i = 0; i < 3; ++i)
			testThat(result[i].size() == PULL_AMOUNT);
		testThat(result[0].type() == INT);
		testThat(result[1].type() == DOUBLE);
		testThat(result[2].type() == BOOL);
		testThat(result[0].at<int32_t>(10) == 2);
		testThat(result[0].at<int32_t>(11) == 2);
		testThat(result[1].at<double>(22) == 2.0);
		testThat(result[1].at<double>(25) == 2.0);
		testThat(result[2].at<bool>(110) == true);
		testThat(result[2].at<bool>(111) == true);

		result = filter.pull();

		testThat(result.size() == 3);
		for (int i = 0; i < 3; ++i)
			testThat(result[i].size() == PULL_AMOUNT / 2);
		testThat(result[0].type() == INT);
		testThat(result[1].type() == DOUBLE);
		testThat(result[2].type() == BOOL);
		testThat(result[0].at<int32_t>(10) == 2);
		testThat(result[0].at<int32_t>(11) == 2);
		testThat(result[1].at<double>(22) == 2.0);
		testThat(result[1].at<double>(25) == 2.0);
		testThat(result[2].at<bool>(110) == true);
		testThat(result[2].at<bool>(111) == true);

		delete server;
	}
	{
		SharedArray<Column> in(8);
		in[0] = Column(INT, 0);
		in[1] = Column(INT, 0);
		in[2] = Column(BOOL, 0);
		in[3] = Column(DOUBLE, 0);
		in[4] = Column(INT, 0);
		in[5] = Column(BOOL, 0);
		in[6] = Column(INT, 0);
		in[7] = Column(DOUBLE, 0);

		testThat(in[3].type() == DOUBLE);

		SharedArray<AggregationData> aggregations(6);
		aggregations[0] = AggregationData(AggregationData::COUNT, 2);
		aggregations[1] = AggregationData(AggregationData::SUM, 0);
		aggregations[2] = AggregationData(AggregationData::COUNT, 5);
		aggregations[3] = AggregationData(AggregationData::SUM, 3);
		aggregations[4] = AggregationData(AggregationData::SUM, 3);
		aggregations[5] = AggregationData(AggregationData::SUM, 1);

		SharedArray<int> summedIntColumns = summedColumns(in, aggregations, INT);

		testThat(summedIntColumns.size() == 2);
		testThat(summedIntColumns[0] == 0);
		testThat(summedIntColumns[1] == 1);

		SharedArray<int> summedDoubleColumns = summedColumns(in, aggregations, DOUBLE);

		testThat(summedDoubleColumns.size() == 1);
		testThat(summedDoubleColumns[0] == 3);

		bool counting = areThereCountColumns(aggregations);
		testThat(counting == true);

		int counts = countColumnsCount(aggregations);
		testThat(counts == 2);

		SharedArray<int> outValueMapping = outValueColumns(in, aggregations);

		testThat(outValueMapping.size() == aggregations.size());
		testThat(outValueMapping[0] == 3);
		testThat(outValueMapping[1] == 1);
		testThat(outValueMapping[2] == 3);
		testThat(outValueMapping[3] == 0);
		testThat(outValueMapping[4] == 0);
		testThat(outValueMapping[5] == 2);
	}
	{
		SharedArray<Column> in(4);
		in[0] = Column(INT, 0);
		in[1] = Column(INT, 0);
		in[2] = Column(BOOL, 0);
		in[3] = Column(DOUBLE, 0);

		SharedArray<int> groupedColumns(2);
		groupedColumns[0] = 0;
		groupedColumns[1] = 2;

		SharedArray<AggregationData> aggregations(3);
		aggregations[0] = AggregationData(AggregationData::SUM, 3);
		aggregations[1] = AggregationData(AggregationData::COUNT, 0);
		aggregations[2] = AggregationData(AggregationData::SUM, 1);

		SharedArray<int> summedIntColumns = summedColumns(in, aggregations, INT);

		testThat(summedIntColumns.size() == 1);
		testThat(summedIntColumns[0] == 1);

		SharedArray<int> summedDoubleColumns = summedColumns(in, aggregations, DOUBLE);

		testThat(summedDoubleColumns.size() == 1);
		testThat(summedDoubleColumns[0] == 3);

		bool counting = areThereCountColumns(aggregations);
		testThat(counting == true);

		int counts = countColumnsCount(aggregations);
		testThat(counts == 1);

		SharedArray<int> outValueMapping = outValueColumns(in, aggregations);

		testThat(outValueMapping.size() == aggregations.size());
		testThat(outValueMapping[0] == 0);
		testThat(outValueMapping[1] == 2);
		testThat(outValueMapping[2] == 1);
	}
	{
		SharedArray<Type> types(1);
		types[0] = BOOL;
		SharedArray<int> columns(1);
		columns[0] = 0;

		Server* server = new OneTwoServer(3 * PULL_AMOUNT + 1);
		testThat(PULL_AMOUNT % 2 == 0);
		OperationNode* scan = new ScanOperationNode(types, columns, server);

		SharedArray<int> groupedColumns(1);
		groupedColumns[0] = 0;
		SharedArray<AggregationData> aggregations(1);
		aggregations[0] = AggregationData(AggregationData::COUNT, 0);

		GroupByOperationNode groupBy(scan, groupedColumns, aggregations);

		SharedArray<Column> result = groupBy.pull();

		testThat(result.size() == 2);
		testThat(result[0].size() == 2);
		testThat(result[1].size() == 2);
		testThat(result[0].type() == BOOL);
		testThat(result[1].type() == INT);

		int smallerValues = (int) result[0].at<bool>(0);
		testThat(result[0].at<bool>(0) != result[0].at<bool>(1));
		testThat(result[1].at<int32_t>(smallerValues) == 3 * PULL_AMOUNT / 2 + 1);
		testThat(result[1].at<int32_t>(1 - smallerValues) == 3 * PULL_AMOUNT / 2);

		delete server;
	}
	{
		SharedArray<Type> types(4);
		types[0] = INT;
		types[1] = INT;
		types[2] = BOOL;
		types[3] = DOUBLE;
		SharedArray<int> columns(4);
		for (int i = 0; i < 4; ++i)
			columns[i] = i;

		Server* server = new OneTwoServer(10 * PULL_AMOUNT);
		testThat(PULL_AMOUNT % 2 == 0);
		OperationNode* scan = new ScanOperationNode(types, columns, server);

		SharedArray<int> groupedColumns(2);
		groupedColumns[0] = 0;
		groupedColumns[1] = 2;
		SharedArray<AggregationData> aggregations(3);
		aggregations[0] = AggregationData(AggregationData::SUM, 3);
		aggregations[1] = AggregationData(AggregationData::COUNT, 0);
		aggregations[2] = AggregationData(AggregationData::SUM, 1);

		GroupByOperationNode groupBy(scan, groupedColumns, aggregations);

		SharedArray<Column> result = groupBy.pull();

		testThat(result.size() == 5);
		for (int i = 0; i < 5; ++i)
			testThat(result[i].size() == 2);
		testThat(result[0].type() == INT);
		testThat(result[1].type() == BOOL);
		testThat(result[2].type() == DOUBLE);
		testThat(result[3].type() == INT);
		testThat(result[4].type() == INT);

		int smallerValues = (result[0].at<int32_t>(0) == 1 ? 0 : 1);
		testThat(result[0].at<int32_t>(smallerValues) == 1);
		testThat(result[0].at<int32_t>(1 - smallerValues) == 2);
		testThat(result[1].at<bool>(smallerValues) == false);
		testThat(result[1].at<bool>(1 - smallerValues) == true);
		testThat(result[2].at<double>(smallerValues) == 5.0 * PULL_AMOUNT);
		testThat(result[2].at<double>(1 - smallerValues) == 10.0 * PULL_AMOUNT);
		testThat(result[3].at<int32_t>(smallerValues) == 5 * PULL_AMOUNT);
		testThat(result[3].at<int32_t>(1 - smallerValues) == 5 * PULL_AMOUNT);

		result = groupBy.pull();

		testThat(result.size() == 5);
		for (int i = 0; i < 5; ++i)
			testThat(result[i].size() == 0);

		delete server;
	}

	cout << "UNIT TESTS PASSED!\n";

	return 0;
}
